package dlab.base


import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}


/**
  * @title: WordCountSparkStreaming
  * @projectName spark-test
  * @description: 使用 netcat 工具向 8888 端口不断的发送数据，通过 SparkStreaming 读取端口数据并统计不同单词出现的次数。
  * @author chengyijian
  * @date 2020/10/27
  */

object wordCount_SparkStreaming {
  def main(args: Array[String]): Unit = {
    // 1、初始化 Spark 配置信息
    val sparkConf: SparkConf = new SparkConf()
      .setMaster("local[*]")
      .set("spark.driver.host", "localhost")
      .setAppName("sparkStreamWordCount")

    // 2、初始化 SparkStreamingContext
    val ssc: StreamingContext = new StreamingContext(sparkConf, Seconds(5))

    val wordAndCountStreams = ssc.socketTextStream("localhost", 8888)
      .flatMap(_.split(" "))
      .map((_,1))
      .reduceByKey(_+_)

    wordAndCountStreams.print()
    ssc.start()
    ssc.awaitTermination()
  }

}